package com.niit.service


import com.niit.utils.SparkUtil
import org.apache.spark.sql._

class MovieDataService {

  //离线分析 ，对电影数据进行分析
  def dataAnalysis(): Unit ={
    val spark: SparkSession = SparkUtil.takeSpark()
    val ds: Dataset[String] = spark.read.textFile("input/rating_100k.data")
    import spark.implicits._ //隐式转换

    //需求：对电影评分数据进行分析，获得电影平均分Top10,要求电影次数大于200

    //处理数据--先获得电影id，评分
    //  196	242	3	881250949
    val movieDF: DataFrame = ds.map(line => { //line  196	242	3	881250949  = [196,242,3,881250949]
      val arrs: Array[String] = line.split("\t")
      var movieId = arrs(1)
      var score = arrs(2).toInt
      (movieId, score) // k - v c
    }).toDF("movieId", "score")

    //利用SQL语句进行分析，获得电影评分分Top10，电影次数大于200
    /*movieDF.createOrReplaceTempView("movie")
    val sql =
        """
          | select movieId,avg(score) avgScore,count(*) counts
          |    from movie
          |    group by movieId
          |    having counts > 200
          |    order by avgScore desc
          |    limit 10
          |
          |""".stripMargin
    spark.sql(sql).show()*/
    //DF.wrire 保存数据


    //利用DSL的方式进行分析
    import org.apache.spark.sql.functions._ //导入DSL的全部函数
    val resDF: Dataset[Row] = movieDF.groupBy("movieId")  //agg 一般是配合groupBy使用，是对分组后的数据进行计算和统计
      .agg(
        count("movieId") as ("counts"),
        avg("score") as ("avgScore")
      ).filter('counts > 200)
      .orderBy('avgScore.desc)
      .limit(10)

    resDF.show()

    //对数据库的操作应该写在dao层
    resDF.write
      .format( "jdbc") //连接MySql JDBC
      .option("url", "jdbc:mysql://node1:3306/BD2") //连接地址 MySQL数据库的版本 8.0+  localhost == 127.0.0.1== 本机IP
      .option("driver", "com.mysql.jdbc.Driver")//数据库驱动  8.0的驱动  com.mysql.cj.jdbc.Driver 5.0：com.mysql.jdbc.Driver
      .option("user", "root")//数据库的用户名
      .option("password", "Niit@123")//数据库的密码
      .option("dbtable", "movie")//数据库表
      .mode(SaveMode.Append)//当该表不存在的时候 会自动创建新表
      .save()//加载，将该表里面的数据封装到DataFrame中

    spark.close()
  }



}
